package com.xzj.admin.resource.analysis.handle.kafka;

import com.xzj.admin.resource.analysis.cache.BaseCache;
import com.xzj.admin.resource.analysis.handle.analysis.pg.MsgAnalysisHandle;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.core.annotation.Order;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * @program: xzj-admin
 * @ClassName: ConsumerHandle
 * @description: kafka消费者
 * @author: XiaZunJun
 * @create: 2024-11-04 16:55
 */
@Component
@Order(value = 20)
@Slf4j
public class ConsumerHandle {

    @Resource
    private MsgAnalysisHandle msgAnalysisHandle;

    @KafkaListener(topics = "#{'${spring.kafka.consumer-topic}'.split(',')}")
    public void kafkaListener(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
        // 判断是否初始化完成，阻塞式
        try {
            // 判断是否初始化完成，阻塞式
            BaseCache.judgeInitFinish();
            log.info("需处理消费数据条数开始：{}",records.size());
            for(ConsumerRecord<String, String> record: records){
                msgAnalysisHandle.analysisKafkaMsg(record);
            }
            log.info("需处理消费数据条数结束：{}",records.size());
        } catch (Exception e) {
            log.error("消息解析失败", e);
        } finally {
            ack.acknowledge();
        }
    }

}
